/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.just.dusin.ooii.component.kj;

import java.sql.SQLException;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.Queue;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.dusin.webcoal.exception.NoReplyException;
import org.just.dusin.ooii.component.Source;
import org.just.dusin.ooii.database.CLCSBean;
import org.just.dusin.ooii.database.DataOf5mBean;
import org.just.dusin.ooii.database.DatabaseConnectionSQL;
import org.just.dusin.ooii.database.QHJCXXBDao;
import org.just.dusin.ooii.database.QHSSSJBean;
import org.just.dusin.ooii.database.QHSSSJDao;
import org.just.dusin.ooii.component.kj.DataPackage;

/**
 *
 * @author Administrator
 */
public class MDatabaseSource extends Source{
    
    private static Logger logger= Logger.getLogger(MDatabaseSource.class.getName());
    private int innerInteval=100;
    private int inteval=1000;
    private DatabaseConnectionSQL dbc=null;
    private QHSSSJDao qhsssjDao =null;
    private QHJCXXBDao qhjcxxbDao = null;
    //private Queue<QHSSSJBean> queue = new LinkedList<QHSSSJBean>();
    private Queue queue = new LinkedList();

    public DatabaseConnectionSQL getDbc() {
        return dbc;
    }

    public void setDbc(DatabaseConnectionSQL dbc) {
        this.dbc = dbc;
    }    
    
    public void setInteval(int inteval) {
        this.inteval = inteval;
    }

    public void setInnerInteval(int innerInteval) {
        this.innerInteval = innerInteval;
    }

    
    @Override
    public void initComponent(Properties props) {
       
    }

    @Override
    public void destroyComponent() throws Exception {
       
    }

    @Override
    public void run() {
        int exceptionOccurred=0;
        while (true) {
            try {
                //dbc = new DatabaseConnectionSQL();
                qhjcxxbDao = new QHJCXXBDao(dbc.getConnection());
                qhsssjDao = new QHSSSJDao(dbc.getConnection());
                dbc.getConnection().setAutoCommit(false);
                qhjcxxbDao.prepareUpload(3);
                List<CLCSBean> lista = qhjcxxbDao.getTops(3);
                qhsssjDao.prepareUpload(10);
                List<QHSSSJBean> list = qhsssjDao.getTops(10);
                queue.addAll(lista);
                queue.addAll(list);
                String msg=null;
                while(!queue.isEmpty()){
                    
                    try{
                        if(exceptionOccurred==1){
                            exceptionOccurred=0;
                            Thread.sleep(innerInteval);
                        }
                        Object mineData = queue.peek();
                        
                        if(mineData instanceof QHSSSJBean){
                            
                            msg = DataPackage.pack((QHSSSJBean)mineData);
                        }else{
                            msg = DataPackage.packCLCS((CLCSBean)mineData);
                        }
                        
                        logger.info("处理前"+msg);
                        sink.process(msg);                        
                    }catch(InterruptedException e){
                        
                    }catch(NoReplyException e){
                        logger.info("无应答，继续发送消息");                        
                        continue;
                    }catch(Exception e){
                        e.printStackTrace();
                        exceptionOccurred=1;
                        logger.warning(e.getMessage());
                        continue;
                    }
                    logger.info("处理下个消息");
                    queue.poll();
                }
                qhsssjDao.finishUpload();
                qhjcxxbDao.finishUpload();
                dbc.getConnection().commit();
                dbc.close();
                Thread.sleep(inteval);
            } catch(InterruptedException e){
              
            }catch(Exception e){
                e.printStackTrace();
                try {
                    dbc.getConnection().rollback();
                } catch (SQLException ex) {
                    ex.printStackTrace();
                    logger.warning(ex.getMessage());
                }
            }finally{
                try {
                    dbc.close();                    
                    Thread.sleep(inteval);
                } catch (InterruptedException ex) {
              
                }
            }            
        }
    }
    
}
